package org.finesys.common.mp.core.transaction;

import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import lombok.Builder;
import lombok.RequiredArgsConstructor;

/**
 * 多线程事务一致性管理 <br>
 * 声明式事务管理无法完成,此时我们只能采用初期的编程式事务管理才行
 */
@Component
@RequiredArgsConstructor
public class MultiplyThreadTransactionManager {

	private final static Logger logger = LoggerFactory.getLogger(MultiplyThreadTransactionManager.class);

	/**
	 * 如果是多数据源的情况下,需要指定具体是哪一个数据源
	 */
	private final DataSource dataSource;

	/**
	 * 执行的是无返回值的任务
	 * @param tasks 异步执行的任务列表
	 * @param executor 异步执行任务需要用到的线程池,考虑到线程池需要隔离,这里强制要求传
	 */
	public void runAsyncButWaitUntilAllDown(List<Runnable> tasks, Executor executor) {
		if (executor == null) {
			throw new IllegalArgumentException("线程池不能为空");
		}
		DataSourceTransactionManager dataSourceTransactionManager = getTransactionManager();
		AtomicBoolean ex = new AtomicBoolean();
		List<CompletableFuture<?>> taskFutureList = new ArrayList<>(tasks.size());
		List<TransactionStatus> transactionStatusList = new ArrayList<>(tasks.size());
		List<TransactionResource> transactionResources = new ArrayList<>(tasks.size());
		tasks.forEach(task -> {
			// 创建任务执行链
			CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
				try {
					// 1、开启新事物
					transactionStatusList.add(openNewTransaction(dataSourceTransactionManager));
					// 2、copy事务资源
					transactionResources.add(TransactionResource.copyTransactionResouces());
					// 3、异步任务执行
					task.run();
				}
				catch (Throwable throwable) {
					// 打印异常信息
					throwable.printStackTrace();
					// 标记异常信息
					ex.set(Boolean.TRUE);
					// 其他任务还没执行的不需要继续执行了
					taskFutureList.forEach(completableFuture -> completableFuture.cancel(true));
				}
			}, executor);
			taskFutureList.add(future);
		});
		// 阻塞直到所有任务全部执行结束---如果有任务被取消,这里会抛出异常,需要捕获
		try {
			CompletableFuture.allOf(taskFutureList.toArray(new CompletableFuture[] {})).get();
		}
		catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
		}
		// 发生了异常
		if (ex.get()) {
			if (logger.isInfoEnabled()) {
				logger.info("发生异常，全部事务回滚");
			}
			for (int i = 0; i < tasks.size(); i++) {
				transactionResources.get(i).autoWiredTransactionResource();
				dataSourceTransactionManager.rollback(transactionStatusList.get(i));
				transactionResources.get(i).removeTransationResource();
			}
		}
		else {
			if (logger.isInfoEnabled()) {
				logger.info("全部事务正常提交");
			}
			for (int i = 0; i < tasks.size(); i++) {
				transactionResources.get(i).autoWiredTransactionResource();
				dataSourceTransactionManager.commit(transactionStatusList.get(i));
				transactionResources.get(i).removeTransationResource();
			}
		}
	}

	private TransactionStatus openNewTransaction(DataSourceTransactionManager dataSourceTransactionManager) {
		DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
		return dataSourceTransactionManager.getTransaction(definition);
	}

	private DataSourceTransactionManager getTransactionManager() {
		return new DataSourceTransactionManager(dataSource);
	}

	@Builder
	public static class TransactionResource {

		// 事务结束后默认会移除集合中的DataSource作为key关联的资源记录
		private Map<Object, Object> resources;

		private Set<TransactionSynchronization> synchronizations;

		private String currentTransactionName;

		private Boolean currentTransactionReadonly;

		private Integer currentTransactionIsolationLevel;

		private Boolean actualTransactionActive;

		public static TransactionResource copyTransactionResouces() {
			return TransactionResource.builder()
				.resources(TransactionSynchronizationManager.getResourceMap())
				.synchronizations(new LinkedHashSet<>())
				.currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName())
				.currentTransactionReadonly(TransactionSynchronizationManager.isCurrentTransactionReadOnly())
				.currentTransactionIsolationLevel(
						TransactionSynchronizationManager.getCurrentTransactionIsolationLevel())
				.actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
				.build();
		}

		public void autoWiredTransactionResource() {
			resources.forEach(TransactionSynchronizationManager::bindResource);
			// 如果需要注册事务监听者,这里记得修改--我们这里不需要,就采用默认负责--spring事务内部默认也是这个值
			TransactionSynchronizationManager.initSynchronization();
			TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);
			TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);
			TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadonly);
			TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);

		}

		public void removeTransationResource() {
			// 事务结束后默认会移除集合中的DataSource作为key关联的资源记录
			// DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错
			resources.keySet().forEach(key -> {
				if (!(key instanceof DataSource)) {
					TransactionSynchronizationManager.unbindResource(key);
				}
			});

		}

	}

}
